package p.ithorns.integration.hazelcast.mqtt;

import com.github.rholder.retry.Retryer;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import p.ithorns.framework.common.utils.JsonUtil;
import p.ithorns.framework.common.utils.RetryUtil;

import java.nio.charset.StandardCharsets;

/**
 * @Author: liuhaitao
 * @Date: 2022/8/17 18:26
 * @Description: MQTT客户端
 */
@Slf4j
public class LocalMqttClient {

    private MqttClient client;

    private MqttClientProps config;

    private IMqttMessageListener listener;

    private final Retryer<Boolean> retryer;

    private LocalMqttClient() {
        retryer = RetryUtil.create(Exception.class);
    }

    /**
     * Holder单例
     */
    private static class Holder {
        private final static LocalMqttClient INSTANCE = new LocalMqttClient();
    }

    public static LocalMqttClient getInstance() {
        return Holder.INSTANCE;
    }


    public boolean isValid() {
        return null != client;
    }

    public void subscribe(final String topic) {
        try {
            retryer.call(() -> {
                client.subscribe(topic, listener);
                log.info("MQTT订阅主题: [{}]", topic);
                return true;
            });
        } catch (Exception e) {
            log.error("主题[{}]订阅失败.", topic, e);
        }
    }

    public void unsubscribe(String topic) {
        try {
            client.unsubscribe(topic);
        } catch (MqttException e) {
            log.error("取消订阅主题[{}]失败", topic, e);
        }
    }

    /**
     * 初始化客户端
     */
    public synchronized void init(
            final MqttClientProps config,
            final MqttMessageDispatcher messageListener
    ) throws MqttException {
        this.config = config;
        this.listener = messageListener;
        String protocol = MqttClientProps.TCP;
        if (config.isHasSsl()) {
            protocol = MqttClientProps.SSL;
        }
        String brokerUrl = protocol + config.getBroker() + ":" + config.getPort();
        this.client = new MqttClient(brokerUrl, "client-" + System.currentTimeMillis());
        // MQTT配置对象
        MqttConnectOptions options = options(config);
        // 设置自动重连, 其它具体参数可以查看MqttConnectOptions
        options.setAutomaticReconnect(true);
        if (!client.isConnected()) {
            @SuppressWarnings("AlibabaAvoidManuallyCreateThread")
            Thread thread = new Thread(() -> {
                try {
                    retryer.call(() -> {
                        client.connect(options);
                        log.info("MQTT客户端：连接服务端[{}]成功", brokerUrl);

                        subscribe(Topics.Data.GATE_RAW.getTopic());
                        return null;
                    });
                } catch (Exception e) {
                    log.error(String.format("MQTT客户端：连接服务器[%s]失败", brokerUrl));
                }
            });
            thread.setDaemon(true);
            thread.start();
        }
    }

    /**
     * 参数配置
     *
     * @param config -
     * @return MqttConnectOptions -
     */
    private MqttConnectOptions options(MqttClientProps config) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setConnectionTimeout(15);
        options.setKeepAliveInterval(config.getKeepalive());
        options.setAutomaticReconnect(true);
        options.setCleanSession(false);
        options.setMaxInflight(config.getMaxInflight());

        if (config.isWithUserNamePass()) {
            String password = config.getPassword();
            if (password != null) {
                options.setPassword(password.toCharArray());
            }
            String username = config.getUsername();
            if (username != null) {
                options.setUserName(username);
            }
        }
        return options;
    }


    /**
     * 发送消息
     *
     * @param topic 主题
     * @param data  消息内容
     */
    public void send(String topic, Object data) {
        try {
            // 转换消息为json字符串
            String json = JsonUtil.toJson(data);
            if (null != json) {
                byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
                client.publish(topic, bytes, this.config.getQos(), false);
            }
        } catch (Exception e) {
            log.error(String.format("MQTT: 主题[%s]发送消息失败", topic));
        }
    }

    public void destroy() throws MqttException {
        if (null != client) {
            client.disconnectForcibly();
            client.close(true);
            log.info("MqttClient closed.");
        }
    }

}